/*
 *  Arrow
 *
 *  Created by David Giovannini on 1/5/07.
 *  Copyright 2007 Software by Jove. All rights reserved.
 *
 */
 
#ifndef SBJ_Arrow_Queue_h
#define SBJ_Arrow_Queue_h

#include "SBJ/Container/NodeQueue.h"
#include "SBJ/Container/Array.h"
#include "SBJ/Interfaces/IExceptionObserver.h"
#include "SBJ/Arrow/IPublisher.h"
#include "SBJ/Arrow/QueueStats.h"

#include <boost/scoped_ptr.hpp>
#include <boost/thread/recursive_mutex.hpp>
#include <boost/thread/condition.hpp>
#include <boost/thread/thread.hpp>
#include <boost/noncopyable.hpp>

#include <hash_map>
#include <vector>

namespace SBJ
{

namespace Arrow
{

class EnqueuePolicy;
class DequeuePolicy;

class Queue : public IRouter, public IPublisher, private boost::noncopyable
{
public:
	typedef boost::shared_ptr< Queue > Ptr;
	typedef NodeQueue< Message > MsgQueue;
	typedef Array< MsgQueue > MsgLists;

	struct Stats
	{
		DequeueStat::Result dequeue;
		EnqueueStat::Result enqueue;
	};
	
	Queue( size_t priorityCount = 1, const IExceptionObserver::Ptr& excObserver = IExceptionObserver::Ptr(),
			EnqueuePolicy* enqueuePolicy = 0, DequeuePolicy* dequeuePolicy = 0 );

	virtual ~Queue( void );

	inline Stats getStats( void ) const;
	
	virtual bool start( void );
	
	virtual bool stop( bool join = true );
	
	void flush( void );
	
	virtual bool addRouter( const IRouter::Ptr& router );
	
	virtual bool removeRouter( const IRouter::Ptr& router );
	
	virtual void onMessage( Message& msg );
	
private:
	typedef Array< size_t > QueueMaxSizes;
	typedef stdext::hash_map< Topic, Message* > KeyedNodes;
	
	bool validityCheck( Message& msg );
	size_t determinePriority( Message& msg );
	Message* boxMessage( Message& msg );
	void enqueueNode( Message* node, size_t priority );

	void threadWork( void );
	bool dequeue( void );

	Message::Priority haveMessage( Message::Priority high, Message::Priority low ) const;
	
	const bool m_hasInterest;
	volatile bool m_started;
	KeyedNodes m_knownNodes;
	MsgLists m_queues;
	QueueMaxSizes m_maxSizes;
	IExceptionObserver::Ptr m_excObserver;
	IRouter::Ptr m_router;

	DequeueStat m_dequeueStat;
	EnqueueStat m_enqueueStat;
	
	boost::scoped_ptr< boost::thread > m_dequeueThread;
	boost::condition m_queueCondition;
	boost::recursive_mutex m_queueGuard;
	boost::recursive_mutex m_collisionGuard;
	Message::Priority m_wakeupPriority;
	
	boost::scoped_ptr< EnqueuePolicy > m_enqueuePolicy;
	boost::scoped_ptr< DequeuePolicy > m_dequeuePolicy;
};

}

}

#include "Queue.inl"

#endif